package com.ibm.cps.message;

import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonParser;
import com.ibm.cps.message.MessageFactory.MessageType;
import com.ibm.util.JSONUtility;
import com.ibm.util.exception.CPSException;

public class DFTMetaMessage extends ProcessorMetaMessage implements
		KryoSerializable {

	int size;
	int step;
	int coefficient;
	String attributename;

	public DFTMetaMessage(String tenantid, String processorId,
			Set<String> parentProcessors, String attributeName, int size,
			int step, int coefficient) {
		this.tenantid = tenantid;
		this.processorid = processorId;
		this.parentprocessors = parentProcessors;
		this.size = size;
		this.step = step;
		this.type = MessageType.DFT;
		this.attributename = attributeName;
		this.coefficient = coefficient;
	}

	/**
	 * json format
	 * tenantid schemaid topicName rulename ruleStr
	 * @param ruleString
	 */
	public DFTMetaMessage(byte[] dftByte) {
		try {
			String dft = new String(dftByte);
			ObjectNode objectNode = JSONUtility.fromObjectString(dft);

			processorid = objectNode.get(MessageFields.PROCESSORID).asText();
			size = objectNode.get(MessageFields.SIZE).asInt();
			step = objectNode.get(MessageFields.STEP).asInt();
			type = MessageType.DFT;
			attributename = objectNode.get(MessageFields.ATTRIBUTENAME)
					.asText();
			coefficient = objectNode.get(MessageFields.COEFFICIENT).asInt();
			ArrayNode arrayNode = (ArrayNode) objectNode
					.get(MessageFields.PARENTPROCESSORS);
			parentprocessors = new HashSet<String>(arrayNode.size());
			for (int i = 0; i < arrayNode.size(); i++) {
				parentprocessors.add(arrayNode.get(i).asText());
			}
		} catch (CPSException e) {
			e.printStackTrace();
		}
	}

	public String getDFTJsonStr() {
		Gson gson = new Gson();
		String json = gson.toJson(this);
		try {
			ObjectNode objectNode = JSONUtility.fromObjectString(json);
			objectNode.remove("type");
			objectNode.remove("deletion");
			objectNode.put("type", "dft");
			return objectNode.toString();
		} catch (CPSException e) {
			e.printStackTrace();
		}
		return null;
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(tenantid);
		output.writeString(processorid);
		output.writeInt(size);
		output.writeInt(step);
		output.writeInt(coefficient);
		output.writeString(attributename);
		kryo.writeObject(output, parentprocessors);
		kryo.writeObject(output, type);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		tenantid = input.readString();
		processorid = input.readString();
		size = input.readInt();
		step = input.readInt();
		coefficient = input.readInt();
		attributename = input.readString();
		// Spark Workaround here
		parentprocessors = kryo.readObject(input, HashSet.class);
		type = kryo.readObject(input, MessageType.class);
	}

	public int getSize() {
		return size;
	}

	public int getStep() {
		return step;
	}

	public int getCoefficient() {
		return coefficient;
	}

	public String getAttributeName() {
		return attributename;
	}

	public static String getCollectionDFT(Collection<DFTMetaMessage> dfts) {
		JsonArray arrayNode = new JsonArray();
		JsonParser parser = new JsonParser();
		for (DFTMetaMessage dft : dfts) {
			arrayNode.add(parser.parse(dft.getDFTJsonStr()));
		}
		return arrayNode.toString();

	}

	@Override
	public void updateProcessorMap(
			HashMap<String, Set<String>> parent2Successors) {
		for (String parent : parentprocessors) {
			if (parent2Successors.get(parent) == null) {
				parent2Successors.put(parent, new HashSet<String>());
			}
			parent2Successors.get(parent).add(processorid);
		}
	}

	public static void main(String[] args) {
		JsonArray arrayNode = new JsonArray();

		System.out.println(arrayNode.toString());
	}
}
